package com.example.service.impl;

import com.example.entity.User;
import com.example.repository.UserRepository;
import com.example.service.KafkaMessageService;
import com.example.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

/**
 * 用户服务实现类
 * 采用Cache-Aside Pattern实现缓存与数据库双写一致性
 */
@Slf4j
@Service
public class UserServiceImpl implements UserService {

    @Autowired
    private UserRepository userRepository;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private KafkaMessageService kafkaMessageService;

    // 缓存key前缀
    private static final String USER_CACHE_PREFIX = "user:";
    // 缓存过期时间（分钟）
    private static final long CACHE_EXPIRE_TIME = 30;

    /**
     * 保存用户信息
     * 策略：先写数据库，再写缓存（Write-Behind）
     * @param user 用户对象
     * @return 保存后的用户对象
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public User saveUser(User user) {
        log.info("保存用户信息到数据库: {}", user);
        
        // 1. 先写数据库
        User savedUser = userRepository.save(user);
        
        // 2. 再写缓存（如果缓存写入失败，不影响主流程）
        try {
            String cacheKey = USER_CACHE_PREFIX + savedUser.getId();
            redisTemplate.opsForValue().set(cacheKey, savedUser, CACHE_EXPIRE_TIME, TimeUnit.MINUTES);
            log.info("用户信息写入缓存成功: key={}", cacheKey);
        } catch (Exception e) {
            log.error("用户信息写入缓存失败: {}", e.getMessage(), e);
            // 缓存写入失败不影响主流程，记录日志并通过Kafka异步处理
            kafkaMessageService.sendCacheFailureMessage(
                String.format("保存用户失败:id=%d,错误信息=%s", savedUser.getId(), e.getMessage()));
        }
        
        return savedUser;
    }

    /**
     * 根据ID获取用户信息
     * 策略：缓存优先，缓存未命中则查询数据库并回填缓存（Cache-Aside Pattern）
     * @param id 用户ID
     * @return 用户对象
     */
    @Override
    public Optional<User> getUserById(Long id) {
        if (id == null) {
            return Optional.empty();
        }

        String cacheKey = USER_CACHE_PREFIX + id;
        
        // 1. 先从缓存中查询
        try {
            User cachedUser = (User) redisTemplate.opsForValue().get(cacheKey);
            if (cachedUser != null) {
                log.info("从缓存中获取用户信息: key={}", cacheKey);
                return Optional.of(cachedUser);
            }
        } catch (Exception e) {
            log.error("从缓存中获取用户信息失败: {}", e.getMessage(), e);
        }

        // 2. 缓存未命中，从数据库查询
        Optional<User> userOptional = userRepository.findById(id);
        if (userOptional.isPresent()) {
            User user = userOptional.get();
            log.info("从数据库中获取用户信息: id={}", id);
            
            // 3. 回填缓存（如果缓存写入失败，不影响主流程）
            try {
                redisTemplate.opsForValue().set(cacheKey, user, CACHE_EXPIRE_TIME, TimeUnit.MINUTES);
                log.info("用户信息回填缓存成功: key={}", cacheKey);
            } catch (Exception e) {
                log.error("用户信息回填缓存失败: {}", e.getMessage(), e);
                // 缓存写入失败不影响主流程，记录日志并通过Kafka异步处理
                kafkaMessageService.sendCacheFailureMessage(
                    String.format("回填用户缓存失败:id=%d,错误信息=%s", id, e.getMessage()));
            }
        }
        
        return userOptional;
    }

    /**
     * 更新用户信息
     * 策略：先更新数据库，再删除缓存（先写数据库后删缓存）
     * 采用"先删缓存再更新数据库"或"先更新数据库再删缓存"都有可能在并发情况下出现不一致
     * 这里采用"先更新数据库再删除缓存" + 延迟双删策略来提高一致性
     * @param id 用户ID
     * @param user 更新的用户信息
     * @return 更新后的用户对象
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public User updateUser(Long id, User user) {
        log.info("更新用户信息: id={}, user={}", id, user);
        
        // 1. 先更新数据库
        user.setId(id);
        User updatedUser = userRepository.save(user);
        
        // 2. 删除缓存（延迟双删策略第一步）
        String cacheKey = USER_CACHE_PREFIX + id;
        try {
            redisTemplate.delete(cacheKey);
            log.info("删除用户缓存: key={}", cacheKey);
        } catch (Exception e) {
            log.error("删除用户缓存失败: {}", e.getMessage(), e);
            // 缓存删除失败，通过Kafka异步处理
            kafkaMessageService.sendCacheFailureMessage(
                String.format("删除用户缓存失败:id=%d,错误信息=%s", id, e.getMessage()));
        }
        
        // 3. 延迟双删策略：在一定时间后再次删除缓存，防止在数据库更新和第一次删除缓存之间有请求将旧数据写入缓存
        // 这里可以使用异步线程或消息队列实现延迟删除
        // 为了简化示例，这里使用简单方式模拟延迟删除
        new Thread(() -> {
            try {
                // 模拟延迟500ms
                Thread.sleep(500);
                redisTemplate.delete(cacheKey);
                log.info("延迟删除用户缓存: key={}", cacheKey);
            } catch (Exception e) {
                log.error("延迟删除用户缓存失败: {}", e.getMessage(), e);
                // 缓存删除失败，通过Kafka异步处理
                kafkaMessageService.sendCacheFailureMessage(
                    String.format("延迟删除用户缓存失败:id=%d,错误信息=%s", id, e.getMessage()));
            }
        }).start();
        
        return updatedUser;
    }

    /**
     * 删除用户
     * 策略：先删除数据库，再删除缓存（保证数据一致性）
     * @param id 用户ID
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void deleteUser(Long id) {
        log.info("删除用户: id={}", id);
        
        // 1. 先删除数据库
        userRepository.deleteById(id);
        
        // 2. 再删除缓存
        String cacheKey = USER_CACHE_PREFIX + id;
        try {
            redisTemplate.delete(cacheKey);
            log.info("删除用户缓存: key={}", cacheKey);
        } catch (Exception e) {
            log.error("删除用户缓存失败: {}", e.getMessage(), e);
            // 缓存删除失败，通过Kafka异步处理
            kafkaMessageService.sendCacheFailureMessage(
                String.format("删除用户缓存失败:id=%d,错误信息=%s", id, e.getMessage()));
        }
    }

    /**
     * 获取所有用户
     * @return 用户列表
     */
    @Override
    public List<User> getAllUsers() {
        return userRepository.findAll();
    }
}